/*
 * BSM6.0
 *
 * Copyright 2012-2013 (C) SINODATA CO., LTD. All Rights Reserved.
 */
package com.sinodata.bsm.center.engine.probe;

import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.log4j.Logger;

import com.sinodata.bsm.center.message.Messenger;
import com.sinodata.bsm.center.service.probe.ProbeService;
import com.sinodata.bsm.center.util.spring.SpringContextHolder;
import com.sinodata.bsm.common.constants.SystemStatus;
import com.sinodata.bsm.common.message.ServiceMessage;
import com.sinodata.bsm.common.vo.Probe;

/**
 * <p>
 * Description: 
 * </p>
 *
 * @author tangli
 * @version 1.0

 * <p>
 * History: 
 *
 * Date                     Author         Version     Description
 * ---------------------------------------------------------------------------------
 * 2012-6-6 上午9:29:33          tangli         1.0         To create
 * </p>
 *
 * @since 
 * @see     
 */
public class ProbeStatusManager extends Thread {
    private static Logger logger = Logger.getLogger(ProbeStatusManager.class);
    private static ProbeStatusManager instance = new ProbeStatusManager();
    private final static Long INTERVAL = 60000L;
    private ProbeService probeService = SpringContextHolder.getBean(ProbeService.class);
    private Set<Long> pollSet = new HashSet<Long>();

    /**
     * 
     */
    private ProbeStatusManager() {
        this.start();
    }

    public static ProbeStatusManager getInstance() {
        return instance;
    }

    /**
     * 通知probe进行重启
     * @param probeId
     */
    public void stopProbe(Long probeId) {
        ServiceMessage message = new ServiceMessage();
        message.setSrvId("SystemService");
        message.setMethod("stopProbe");
        message.setParams(new Object[] {});
        try {
            Messenger.getInstance().sendAsynMessage(probeId, message);
        } catch (Exception e) {
            logger.error("停止[" + probeId + "]失败", e);
        }
    }

    /**
     * 先通知运行的probe进行重启
     * 在进行probe的轮询
     */
    public void run() {
        Collections.synchronizedSet(pollSet);
        Long time = System.currentTimeMillis();
        List<Probe> probes = probeService.findAllProbe();
        for (Probe probe : probes) {
            stopProbe(probe.getId());
        }
        logger.info("[TIP]>>>>>Start To Poll Probe.");
        while (true) {
            probes = probeService.findAllProbe();
            Long startTime = System.currentTimeMillis();
            for (Probe probe : probes) {
                if (!pollSet.contains(probe.getId())) {
                    pollSet.add(probe.getId());
                    new ProbePollThread(probe).start();
                }
            }
            time = System.currentTimeMillis() - startTime;
            if (time < INTERVAL) {
                try {
                    Thread.sleep(INTERVAL - time);
                } catch (InterruptedException e) {
                }
            }
        }
    }

    private void checkProbe(Probe probe) {
        // logger.info("轮询probe:" + probe.getId());
        ServiceMessage message = new ServiceMessage();
        message.setSrvId("SystemService");
        message.setMethod("getProbeStatus");
        message.setParams(new Object[] {});
        Object obj = null;
        String[] resp = null;
        try {
            obj = Messenger.getInstance().sendSyncMessage(probe.getId(), message);
        } catch (Exception e) {
            logger.error("与Probe[" + probe.getId() + "]的心跳失败", e);
        }
        if (obj != null && obj instanceof String[]) {
            resp = (String[]) obj;
            int taskSize = Integer.parseInt(resp[0]);
            probe.setTaskCount(taskSize);
            probe.setActiveDate(new Date());
            probe.setStatus(SystemStatus.STATUS_RUN);
            probeService.updateProbe(probe);
        } else {
            probe.setStatus(SystemStatus.STATUS_DOWN);
            probeService.updateProbe(probe);
        }
    }

    class ProbePollThread extends Thread {
        private Probe probe;

        public ProbePollThread(Probe probe) {
            this.probe = probe;
        }

        @Override
        public void run() {
            try {
                checkProbe(probe);
            } catch (Exception e) {
                logger.error(e);
            }
            pollSet.remove(probe.getId());
        }

    }
}
